package com.joysuch.wwyt.streampusher;

import com.google.common.base.Strings;
import com.joysuch.wwyt.core.constant.Constants;
import com.joysuch.wwyt.streampusher.exception.PushException;
import com.joysuch.wwyt.streampusher.publisher.StreamPusher;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 直播流管理器
 * 1、使用addPublisher方法开启流，
 * 2、要想让流一直正常工作，需要不间断的调用breath方法，如果超过DEAD_WITHOUT_BREATH_TIME_MILLISECONDS时间没
 * 调用breath方法，则这个流自动关闭。（此逻辑的作用在于在客户端任何异常或不规范情况下没调用关闭流，则流能自动关闭）
 * 3、使用close接口关闭一个流
 *
 * @author qingzhou
 * 2017-11-08 12:10
 */
@Slf4j
@Component
public class StreamPusherManager {

    private static final Logger logger = LoggerFactory.getLogger(StreamPusherManager.class);

    /**
     * 多久不呼吸就死亡
     */
    private static final long DEAD_WITHOUT_BREATH_TIME_MILLISECONDS = 15 * 1000;

    /**
     * 每个CPU核心最大处理几个视频编解码
     */
    private static final int CODEC_SIZE_PER_PROCESSOR = 5;

    /**
     * 最大并发量，防止开流太多服务器挂掉
     */
    private int maxConcurrency;

    /**
     * 正在推送的视频流
     */
    private Map<String, StreamPusherWrapper> pubs;

    /**
     * 呼吸检测线程池
     */
    private ScheduledThreadPoolExecutor executor;

    /**
     * 是否已经关闭了
     */
    private volatile boolean shutdown;

    /**
     * 应用名称，防止同台服务器安装多个应用程序，相互之间进程误杀
     */
    private String appName;

    public StreamPusherManager() throws PushException {
       // init(Runtime.getRuntime().availableProcessors(), "wwyt");
    }

    public void init(int maxConcurrency, String appName) throws PushException {
        if (maxConcurrency < 1) {
            throw new PushException("推流并发数不能小于1");
        }
        if (maxConcurrency > Runtime.getRuntime().availableProcessors() * CODEC_SIZE_PER_PROCESSOR) {
            throw new PushException("推流并发数不能大于CPU核心数 * " + CODEC_SIZE_PER_PROCESSOR);
        }
        if (Strings.isNullOrEmpty(appName)) {
            throw new PushException("应用程序名称不能为空");
        }
        this.maxConcurrency = maxConcurrency;
        this.appName = appName;
        this.pubs = new HashMap<>(maxConcurrency);
        executor = new ScheduledThreadPoolExecutor(maxConcurrency + 1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "StreamPusherManager-Worker");
            }
        });
        executor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                checkBreath();
            }
        }, 5, 1, TimeUnit.SECONDS);
        executor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                checkFFMPEGProcessLeak();
            }
        }, 1, 1, TimeUnit.MINUTES);
        this.shutdown = false;
    }

    @Autowired
    private JsStreamPusherRedisPublisher jsStreamPusherRedisPublisher;

    /**
     * 新增多个推流器
     *
     * @param list 推流器列表
     * @return 返回每个推流器是否成功
     * @throws PushException 推流异常
     */
    public Map<String, Boolean> addPublisher(List<StreamPusher> list) throws PushException {
        Map<String, Boolean> exeRes = new HashMap<>(list.size());
        List<Callable<Void>> callbacks = new ArrayList<>(list.size());
        for (final StreamPusher sp : list) {
            callbacks.add(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    addPublisher(sp);
                    return null;
                }
            });
        }
        try {
            List<Future<Void>> futures = executor.invokeAll(callbacks, 15, TimeUnit.SECONDS);
            for (int i = 0; i < futures.size(); i++) {
                String identify = list.get(i).identify();
                try {
                    Future<Void> future = futures.get(i);
                    if (future != null) {
                        future.get();
                        exeRes.put(identify, true);
                    } else {
                        exeRes.put(identify, false);
                    }
                } catch (ExecutionException e) {
                    logger.info("启动推流异常:", e);
                    exeRes.put(identify, false);
                } catch (CancellationException e) {
                    logger.info("启动推流超时");
                    exeRes.put(identify, false);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return exeRes;
    }

    /**
     * 新增一个推流器
     *
     * @param push 推流器
     */
    private void addPublisher(StreamPusher push) throws PushException {
        if (this.shutdown) {
            throw new PushException("视频流管理器已经关闭");
        }
        if (push == null) {
            throw new PushException("发布器不能为空");
        }
        String identify = push.identify();
        if (identify == null || identify.trim().length() == 0) {
            throw new PushException("发布器标识符不能为空");
        }
        //反射注入appName属性
        try {
            injectAppName(push);
        } catch (NoSuchFieldException e) {
            //ignore
        } catch (IllegalAccessException e) {
            throw new PushException("注入应用名称异常");
        }
        identify = identify.trim();
        StreamPusherWrapper wrapper = new StreamPusherWrapper(push);
        boolean isAdd = false;
        synchronized (pubs) {
            System.err.println(identify + " sync1 start...");
            if (!pubs.containsKey(identify)) {
                if (pubs.size() >= maxConcurrency) {
                    throw new PushException(identify + " 同时推流并发不能大于: " + maxConcurrency);
                }
                isAdd = true;
            } else {
                logger.info("stream[" + identify + "] already existed, refreshing breath...");
                pubs.get(identify).brease();
            }
            System.err.println(identify + " sync1 end...");
        }
        //耗时操作从同步代码块拿出来
        if (isAdd) {
            logger.info("try adding one stream[" + identify + "] opening...");
            wrapper.getPublisher().open();
            setRedisBreathValue(identify);
            synchronized (pubs) {
                System.err.println(identify + " sync2 start...");
                if (pubs.size() >= maxConcurrency) {
                    wrapper.getPublisher().close();
                    throw new PushException(identify + " 同时推流并发不能大于: " + maxConcurrency);
                }
                logger.info("real adding one stream[" + identify + "]...");
                pubs.put(identify, wrapper);
                wrapper.brease();
                System.err.println(identify + " sync2 end...");
            }
        }
    }

    private void injectAppName(StreamPusher push) throws NoSuchFieldException, IllegalAccessException {
        Class<?> c = push.getClass();
        Field appName = c.getDeclaredField("appName");
        if (appName != null) {
            appName.setAccessible(true);
            appName.set(push, this.appName);
        }
    }

    /**
     * 关闭所有流
     */
    public void shutdown() {
        this.shutdown = true;
        this.executor.shutdown();
    }

    /**
     * 维持呼吸
     */
    public void breath(String identify) throws PushException {
//        String redisIdentifyValue = (String) jsStreamPusherRedisPublisher.getRedissonClient().getBucket(PodConstant.STREAM_BREATH_REDIS_PRE + "_" + identify).get();
//        if (redisIdentifyValue == null) {
//            throw new PushException(identify + " 流不存在");
//        }

        jsStreamPusherRedisPublisher.publish("breath" + "_" + identify);
    }

    public void breathFromOtherServerRequest(String identify) {
        StreamPusherWrapper wrapper = pubs.get(identify);
        if (wrapper == null) {
            return;
        }
        setRedisBreathValue(identify);
        wrapper.brease();
        log.info("stream breath sync identify {}", identify);
    }

    private void setRedisBreathValue(String identify) {
        jsStreamPusherRedisPublisher.getRedissonClient().getBucket(Constants.STREAM_BREATH_REDIS_PRE + "_" + identify).set(identify);
        jsStreamPusherRedisPublisher.getRedissonClient().getBucket(Constants.STREAM_BREATH_REDIS_PRE + "_" + identify).expire(10, TimeUnit.SECONDS);
    }

    /**
     * 手动关闭
     *
     * @param identify
     */
    public void close(String identify) {
        jsStreamPusherRedisPublisher.publish("close" + "_" + identify);
        jsStreamPusherRedisPublisher.getRedissonClient().getBucket(Constants.STREAM_BREATH_REDIS_PRE + "_" + identify).delete();
    }

    public void closeFromOtherServerRequest(String identify) {
        synchronized (pubs) {
            StreamPusherWrapper wrapper = pubs.get(identify);
            if (wrapper != null) {
                try {
                    wrapper.getPublisher().close();
                    log.info("stream close sync identify {}", identify);
                } catch (PushException e) {
                    e.printStackTrace();
                }
                pubs.remove(identify);
            }
        }

    }

    /**
     * 返回所有存在的流的状态
     */
    public Map<String, Boolean> allStreams() {
        synchronized (pubs) {
            Set<String> identifies = pubs.keySet();
            Map<String, Boolean> map = new HashMap<>(identifies.size());
            for (String identify : identifies) {
                StreamPusherWrapper wrapper = pubs.get(identify);
                if (wrapper != null) {
                    map.put(identify, !wrapper.getPublisher().isClosed());
                }
            }
            return map;
        }
    }

    private void checkBreath() {
        synchronized (pubs) {
            try {
                if (pubs.size() > 0) {
                    Iterator<Map.Entry<String, StreamPusherWrapper>> iterator = pubs.entrySet().iterator();
                    while (iterator.hasNext()) {
                        Map.Entry<String, StreamPusherWrapper> next = iterator.next();
                        StreamPusherWrapper val = next.getValue();
                        //太久没心跳（前端关闭了播放）或者程序本身关闭了
                        if (val.getLastBreathTime() + DEAD_WITHOUT_BREATH_TIME_MILLISECONDS < System.currentTimeMillis()
                                || val.getPublisher().isClosed()) {
                            try {
                                val.getPublisher().close();
                                iterator.remove();
                                logger.info("removing dead stream: " + val.getPublisher().identify());
                            } catch (PushException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }
            } catch (Exception e) {
                logger.error("schedule exception:", e);
            }
        }
    }

    private static final Pattern IDENTIFY_PATTERN = Pattern.compile("seekcy.*ffmpeg.*rtmp.*pusher/(.*)");
    private static final Pattern PID_PATTERN = Pattern.compile("seekcy\\s*(\\d+)\\s*.*");

    /**
     * 判断进程泄露
     */
    private void checkFFMPEGProcessLeak() {
//        try {
////            logger.info("check ffmpeg process leak...");
//            List<String> process = ShellScriptExecutor.executeShell("ps aux | grep ffmpeg | grep -v grep", null);
//            if (process != null && process.size() > 0) {
//                for (String cmd : process) {
//                    try {
//                        Matcher identifyMatcher = IDENTIFY_PATTERN.matcher(cmd);
//                        if (identifyMatcher.find()) {
//                            String identifyWithSuffix = identifyMatcher.group(1);
//                            synchronized (pubs) {
//                                logger.info(pubs.keySet().toString());
//                                if (identifyWithSuffix.endsWith("_" + appName)) {
//                                    String identify = identifyWithSuffix.replace("_" + appName, "");
//                                    //只针对相同appName的进程杀掉
//                                    if (!pubs.containsKey(identify)) {
//                                        logger.info("try killing : " + identify);
//                                        //进程不在当前流列表里，杀掉
//                                        Matcher pidMatcher = PID_PATTERN.matcher(cmd);
//                                        if (pidMatcher.find()) {
//                                            String pid = pidMatcher.group(1);
//                                            String kill = "kill -9 " + pid;
//                                            logger.info("do killing : " + identify);
//                                            ShellScriptExecutor.executeShell(kill, null);
//                                        }
//                                    }
//                                }
//                            }
//                        }
//                    } catch (Exception e) {
//                        logger.error("checkFFMPEGProcessLeak for process [" + cmd + "] exception:", e);
//                    }
//                }
//            }
//        } catch (Exception e) {
//            logger.info("checkFFMPEGProcessLeak exception:", e);
//        }
    }

    public boolean validatorHistoryNotExisted() {
        synchronized (pubs) {
            Set<String> identifies = pubs.keySet();
            for (String identify : identifies) {
                String[] ipArr = identify.split("_");
                //历史流格式 : 192.168.0.196_9096_20180227095130-20180227105130
                //实时流格式 : 192.168.0.196_9096
                if (ipArr.length != 2) {
                    //说明含有时间段信息，是历史视频
                    return false;
                }
            }
        }
        return true;
    }

}
